package org.project.pvuv

import org.FlinkStreamApp
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.project.bean.UserBebavior
import org.apache.flink.streaming.api.scala._
/**
 * description ：计算 pv
 * author      ：剧情再美终是戏
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/26 9:16
 * modified By ：
 * version:    : 1.0
 */
object Pv extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 读取源数据
    val source = environment.readTextFile("F:\\Word\\idea\\bigdata\\Flink0830s\\src\\main\\resources\\UserBehavior.csv")

    // 计算 pv
    source
      .map(line => {
        val splits = line.split(",")
        UserBebavior(splits(0).toLong, splits(1).toLong, splits(2).toInt, splits(3), splits(4).toLong * 1000)
      })
      .filter(_.behavior == "pv")
      .assignAscendingTimestamps(_.timestamp)
      .map(_ => ("pv", 1))
      .keyBy(_._1)
      .timeWindow(Time.minutes(60), Time.minutes(5))
      .aggregate(new MyAgg(), new FullWindow)
      .print
  }

  class FullWindow extends ProcessWindowFunction[Long, String, String, TimeWindow] {
    override def process(key: String, context: Context, elements: Iterable[Long], out: Collector[String]) = {
      out.collect(s"窗口时间 ${context.window.getEnd} pv = ${elements.iterator.next()}")
    }
  }

  class MyAgg extends AggregateFunction[(String, Int), Long, Long] {

    override def add(value: (String, Int), accumulator: Long) = accumulator + 1

    override def getResult(accumulator: Long) = accumulator

    override def merge(a: Long, b: Long) = a + b

    override def createAccumulator() = 0
  }


}
